// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//
#include <pollux/functions/prestosql/aggregates/covariance_aggregates.h>
#include <pollux/exec/aggregate.h>
#include <pollux/expression/function_signature.h>
#include <pollux/functions/prestosql/aggregates/aggregate_names.h>
#include <pollux/vector/decoded_vector.h>
#include <pollux/vector/flat_vector.h>

namespace kumo::pollux::aggregate::prestosql {
    namespace {
        // Indices into RowType representing intermediate results of covar_samp and
        // covar_pop. Columns appear in alphabetical order.
        struct CovarIndices {
            int32_t count;
            int32_t meanX;
            int32_t meanY;
            int32_t c2;
        };

        constexpr CovarIndices kCovarIndices{1, 2, 3, 0};

        // Indices into RowType representing intermediate results of corr. Columns
        // appear in alphabetical order.
        struct CorrIndices : public CovarIndices {
            int32_t m2X;
            int32_t m2Y;
        };

        constexpr CorrIndices kCorrIndices{{1, 4, 5, 0}, 2, 3};

        // Indices into RowType representing intermediate results of regr_slope and
        // regr_intercept.
        struct RegrIndices : public CovarIndices {
            int32_t m2X;
            int32_t m2Y;
        };

        constexpr RegrIndices kRegrIndices{{1, 3, 4, 0}, 2, 5};

        struct CovarAccumulator {
            int64_t count() const {
                return count_;
            }

            double meanX() const {
                return meanX_;
            }

            double meanY() const {
                return meanY_;
            }

            double c2() const {
                return c2_;
            }

            void update(double x, double y) {
                count_ += 1;
                double deltaX = x - meanX();
                meanX_ += deltaX / count();
                double deltaY = y - meanY();
                meanY_ += deltaY / count();
                c2_ += deltaX * (y - meanY());
            }

            void merge(
                int64_t countOther,
                double meanXOther,
                double meanYOther,
                double c2Other) {
                if (countOther == 0) {
                    return;
                }
                if (count_ == 0) {
                    count_ = countOther;
                    meanX_ = meanXOther;
                    meanY_ = meanYOther;
                    c2_ = c2Other;
                    return;
                }

                int64_t newCount = countOther + count();
                double deltaMeanX = meanXOther - meanX();
                double deltaMeanY = meanYOther - meanY();
                c2_ += c2Other +
                        deltaMeanX * deltaMeanY * count() * countOther /
                        static_cast<double>(newCount);
                meanX_ += deltaMeanX * countOther / static_cast<double>(newCount);
                meanY_ += deltaMeanY * countOther / static_cast<double>(newCount);
                count_ = newCount;
            }

        private:
            int64_t count_{0};
            double meanX_{0};
            double meanY_{0};
            double c2_{0};
        };

        struct CovarSampResultAccessor {
            static bool hasResult(const CovarAccumulator &accumulator) {
                return accumulator.count() > 1;
            }

            static double result(const CovarAccumulator &accumulator) {
                return accumulator.c2() / (accumulator.count() - 1);
            }
        };

        struct CovarPopResultAccessor {
            static bool hasResult(const CovarAccumulator &accumulator) {
                return accumulator.count() > 0;
            }

            static double result(const CovarAccumulator &accumulator) {
                return accumulator.c2() / accumulator.count();
            }
        };

        template<typename T>
        SimpleVector<T> *asSimpleVector(
            const RowVector *row_vector,
            int32_t childIndex) {
            return row_vector->childAt(childIndex)->as<SimpleVector<T> >();
        }

        class CovarIntermediateInput {
        public:
            explicit CovarIntermediateInput(
                const RowVector *row_vector,
                const CovarIndices &indices = kCovarIndices)
                : count_{asSimpleVector<int64_t>(row_vector, indices.count)},
                  meanX_{asSimpleVector<double>(row_vector, indices.meanX)},
                  meanY_{asSimpleVector<double>(row_vector, indices.meanY)},
                  c2_{asSimpleVector<double>(row_vector, indices.c2)} {
            }

            void mergeInto(CovarAccumulator &accumulator, vector_size_t row) {
                accumulator.merge(
                    count_->value_at(row),
                    meanX_->value_at(row),
                    meanY_->value_at(row),
                    c2_->value_at(row));
            }

        protected:
            SimpleVector<int64_t> *count_;
            SimpleVector<double> *meanX_;
            SimpleVector<double> *meanY_;
            SimpleVector<double> *c2_;
        };

        template<typename T>
        T *mutableRawValues(const RowVector *row_vector, int32_t childIndex) {
            return row_vector->childAt(childIndex)
                    ->as<FlatVector<T> >()
                    ->mutableRawValues();
        }

        class CovarIntermediateResult {
        public:
            explicit CovarIntermediateResult(
                const RowVector *row_vector,
                const CovarIndices &indices = kCovarIndices)
                : count_{mutableRawValues<int64_t>(row_vector, indices.count)},
                  meanX_{mutableRawValues<double>(row_vector, indices.meanX)},
                  meanY_{mutableRawValues<double>(row_vector, indices.meanY)},
                  c2_{mutableRawValues<double>(row_vector, indices.c2)} {
            }

            static std::string type() {
                return "row(double,bigint,double,double)";
            }

            void set(vector_size_t row, const CovarAccumulator &accumulator) {
                count_[row] = accumulator.count();
                meanX_[row] = accumulator.meanX();
                meanY_[row] = accumulator.meanY();
                c2_[row] = accumulator.c2();
            }

        private:
            int64_t *count_;
            double *meanX_;
            double *meanY_;
            double *c2_;
        };

        struct CorrAccumulator : public CovarAccumulator {
            double m2X() const {
                return m2X_;
            }

            double m2Y() const {
                return m2Y_;
            }

            void update(double x, double y) {
                double oldMeanX = meanX();
                double oldMeanY = meanY();
                CovarAccumulator::update(x, y);

                m2X_ += (x - oldMeanX) * (x - meanX());
                m2Y_ += (y - oldMeanY) * (y - meanY());
            }

            void merge(
                int64_t countOther,
                double meanXOther,
                double meanYOther,
                double c2Other,
                double m2XOther,
                double m2YOther) {
                if (countOther == 0) {
                    return;
                }

                if (count() == 0) {
                    m2X_ = m2XOther;
                    m2Y_ = m2YOther;
                } else {
                    auto k = 1.0 * count() / (count() + countOther) * countOther;
                    m2X_ += m2XOther + k * std::pow(meanX() - meanXOther, 2);
                    m2Y_ += m2YOther + k * std::pow(meanY() - meanYOther, 2);
                }

                CovarAccumulator::merge(countOther, meanXOther, meanYOther, c2Other);
            }

        private:
            double m2X_{0};
            double m2Y_{0};
        };

        struct CorrResultAccessor {
            static bool hasResult(const CorrAccumulator &accumulator) {
                return !std::isnan(result(accumulator));
            }

            static double result(const CorrAccumulator &accumulator) {
                double stddevX = std::sqrt(accumulator.m2X());
                double stddevY = std::sqrt(accumulator.m2Y());
                return accumulator.c2() / stddevX / stddevY;
            }
        };

        class CorrIntermediateInput : public CovarIntermediateInput {
        public:
            explicit CorrIntermediateInput(const RowVector *row_vector)
                : CovarIntermediateInput(row_vector, kCorrIndices),
                  m2X_{asSimpleVector<double>(row_vector, kCorrIndices.m2X)},
                  m2Y_{asSimpleVector<double>(row_vector, kCorrIndices.m2Y)} {
            }

            void mergeInto(CorrAccumulator &accumulator, vector_size_t row) {
                accumulator.merge(
                    count_->value_at(row),
                    meanX_->value_at(row),
                    meanY_->value_at(row),
                    c2_->value_at(row),
                    m2X_->value_at(row),
                    m2Y_->value_at(row));
            }

        private:
            SimpleVector<double> *m2X_;
            SimpleVector<double> *m2Y_;
        };

        class CorrIntermediateResult : public CovarIntermediateResult {
        public:
            explicit CorrIntermediateResult(const RowVector *row_vector)
                : CovarIntermediateResult(row_vector, kCorrIndices),
                  m2X_{mutableRawValues<double>(row_vector, kCorrIndices.m2X)},
                  m2Y_{mutableRawValues<double>(row_vector, kCorrIndices.m2Y)} {
            }

            static std::string type() {
                return "row(double,bigint,double,double,double,double)";
            }

            void set(vector_size_t row, const CorrAccumulator &accumulator) {
                CovarIntermediateResult::set(row, accumulator);
                m2X_[row] = accumulator.m2X();
                m2Y_[row] = accumulator.m2Y();
            }

        private:
            double *m2X_;
            double *m2Y_;
        };

        struct RegrAccumulator : public CovarAccumulator {
            double m2X() const {
                return m2X_;
            }

            void update(double x, double y) {
                double oldMeanX = meanX();
                CovarAccumulator::update(x, y);

                m2X_ += (x - oldMeanX) * (x - meanX());
            }

            void merge(
                int64_t countOther,
                double meanXOther,
                double meanYOther,
                double c2Other,
                double m2XOther) {
                if (countOther == 0) {
                    return;
                }
                if (count() == 0) {
                    m2X_ = m2XOther;
                } else {
                    m2X_ += m2XOther +
                            1.0 * count() / (count() + countOther) * countOther *
                            std::pow(meanX() - meanXOther, 2);
                }
                CovarAccumulator::merge(countOther, meanXOther, meanYOther, c2Other);
            }

        protected:
            double m2X_{0};
        };

        struct ExtendedRegrAccumulator : public RegrAccumulator {
            double m2Y() const {
                return m2Y_;
            }

            void update(double x, double y) {
                double oldMeanY = meanY();
                RegrAccumulator::update(x, y);
                m2Y_ += (y - oldMeanY) * (y - meanY());
            }

            void merge(
                int64_t countOther,
                double meanXOther,
                double meanYOther,
                double c2Other,
                double m2XOther,
                double m2YOther) {
                if (countOther == 0) {
                    return;
                }
                if (count() == 0) {
                    m2X_ = m2XOther;
                    m2Y_ = m2YOther;
                } else {
                    m2X_ += m2XOther +
                            1.0 * count() / (count() + countOther) * countOther *
                            std::pow(meanX() - meanXOther, 2);

                    m2Y_ += m2YOther +
                            1.0 * count() / (count() + countOther) * countOther *
                            std::pow(meanY() - meanYOther, 2);
                }
                CovarAccumulator::merge(countOther, meanXOther, meanYOther, c2Other);
            }

        private:
            double m2Y_{0};
        };

        struct RegrCountResultAccessor {
            static bool hasResult(const ExtendedRegrAccumulator &accumulator) {
                return accumulator.count() > 0;
            }

            static double result(const ExtendedRegrAccumulator &accumulator) {
                return accumulator.count();
            }
        };

        struct RegrAvgyResultAccessor {
            static bool hasResult(const ExtendedRegrAccumulator &accumulator) {
                return accumulator.count() > 0;
            }

            static double result(const ExtendedRegrAccumulator &accumulator) {
                return accumulator.meanY();
            }
        };

        struct RegrAvgxResultAccessor {
            static bool hasResult(const ExtendedRegrAccumulator &accumulator) {
                return accumulator.count() > 0;
            }

            static double result(const ExtendedRegrAccumulator &accumulator) {
                return accumulator.meanX();
            }
        };

        struct RegrSxyResultAccessor {
            static bool hasResult(const ExtendedRegrAccumulator &accumulator) {
                return accumulator.count() > 0;
            }

            static double result(const ExtendedRegrAccumulator &accumulator) {
                return accumulator.c2();
            }
        };

        struct RegrR2ResultAccessor {
            static bool hasResult(const ExtendedRegrAccumulator &accumulator) {
                return accumulator.m2X() != 0;
            }

            static double result(const ExtendedRegrAccumulator &accumulator) {
                if (accumulator.m2X() != 0 && accumulator.m2Y() == 0) {
                    return 1;
                }
                return std::pow(accumulator.c2(), 2) /
                       (accumulator.m2X() * accumulator.m2Y());
            }
        };

        struct RegrSyyResultAccessor {
            static bool hasResult(const ExtendedRegrAccumulator &accumulator) {
                return accumulator.count() > 0;
            }

            static double result(const ExtendedRegrAccumulator &accumulator) {
                return accumulator.m2Y();
            }
        };

        struct RegrSxxResultAccessor {
            static bool hasResult(const ExtendedRegrAccumulator &accumulator) {
                return accumulator.count() > 0;
            }

            static double result(const ExtendedRegrAccumulator &accumulator) {
                return accumulator.m2X();
            }
        };

        struct RegrSlopeResultAccessor {
            static bool hasResult(const RegrAccumulator &accumulator) {
                return !std::isnan(result(accumulator));
            }

            static double result(const RegrAccumulator &accumulator) {
                return accumulator.c2() / accumulator.m2X();
            }
        };

        struct RegrInterceptResultAccessor {
            static bool hasResult(const RegrAccumulator &accumulator) {
                return !std::isnan(result(accumulator));
            }

            static double result(const RegrAccumulator &accumulator) {
                double slope = RegrSlopeResultAccessor::result(accumulator);
                return accumulator.meanY() - slope * accumulator.meanX();
            }
        };

        class RegrIntermediateInput : public CovarIntermediateInput {
        public:
            explicit RegrIntermediateInput(const RowVector *row_vector)
                : CovarIntermediateInput(row_vector, kRegrIndices),
                  m2X_{asSimpleVector<double>(row_vector, kRegrIndices.m2X)} {
            }

            void mergeInto(RegrAccumulator &accumulator, vector_size_t row) {
                accumulator.merge(
                    count_->value_at(row),
                    meanX_->value_at(row),
                    meanY_->value_at(row),
                    c2_->value_at(row),
                    m2X_->value_at(row));
            }

        protected:
            SimpleVector<double> *m2X_;
        };

        class ExtendedRegrIntermediateInput : public RegrIntermediateInput {
        public:
            explicit ExtendedRegrIntermediateInput(const RowVector *row_vector)
                : RegrIntermediateInput(row_vector),
                  m2Y_{asSimpleVector<double>(row_vector, kRegrIndices.m2Y)} {
            }

            void mergeInto(ExtendedRegrAccumulator &accumulator, vector_size_t row) {
                accumulator.merge(
                    count_->value_at(row),
                    meanX_->value_at(row),
                    meanY_->value_at(row),
                    c2_->value_at(row),
                    m2X_->value_at(row),
                    m2Y_->value_at(row));
            }

        private:
            SimpleVector<double> *m2Y_;
        };

        class RegrIntermediateResult : public CovarIntermediateResult {
        public:
            explicit RegrIntermediateResult(const RowVector *row_vector)
                : CovarIntermediateResult(row_vector, kRegrIndices),
                  m2X_{mutableRawValues<double>(row_vector, kRegrIndices.m2X)} {
            }

            static std::string type() {
                return "row(double,bigint,double,double,double)";
            }

            void set(vector_size_t row, const RegrAccumulator &accumulator) {
                CovarIntermediateResult::set(row, accumulator);
                m2X_[row] = accumulator.m2X();
            }

        private:
            double *m2X_;
        };

        class ExtendedRegrIntermediateResult : public RegrIntermediateResult {
        public:
            explicit ExtendedRegrIntermediateResult(const RowVector *row_vector)
                : RegrIntermediateResult(row_vector),
                  m2Y_{mutableRawValues<double>(row_vector, kRegrIndices.m2Y)} {
            }

            static std::string type() {
                return "row(double,bigint,double,double,double,double)";
            }

            void set(vector_size_t row, const ExtendedRegrAccumulator &accumulator) {
                RegrIntermediateResult::set(row, accumulator);
                m2Y_[row] = accumulator.m2Y();
            }

        private:
            double *m2Y_;
        };

        // @tparam T Type of the raw input and final result. Can be double or float.
        template<
            typename T,
            typename TAccumulator,
            typename TIntermediateInput,
            typename TIntermediateResult,
            typename TResultAccessor>
        class CovarianceAggregate : public exec::Aggregate {
        public:
            explicit CovarianceAggregate(TypePtr resultType)
                : exec::Aggregate(resultType) {
            }

            int32_t accumulatorFixedWidthSize() const override {
                return sizeof(TAccumulator);
            }

            void addRawInput(
                char **groups,
                const SelectivityVector &rows,
                const std::vector<VectorPtr> &args,
                bool /* mayPushdown */) override {
                if constexpr (
                    std::is_same_v<TAccumulator, RegrAccumulator> ||
                    std::is_same_v<TAccumulator, ExtendedRegrAccumulator>) {
                    // The args order of linear regression function is (y, x), so we need to
                    // swap the order
                    decodedX_.decode(*args[1], rows);
                    decodedY_.decode(*args[0], rows);
                } else {
                    decodedX_.decode(*args[0], rows);
                    decodedY_.decode(*args[1], rows);
                }

                rows.applyToSelected([&](auto row) {
                    if (decodedX_.is_null_at(row) || decodedY_.is_null_at(row)) {
                        return;
                    }
                    auto *group = groups[row];
                    exec::Aggregate::clearNull(group);
                    accumulator(group)->update(
                        decodedX_.value_at<T>(row), decodedY_.value_at<T>(row));
                });
            }

            void addIntermediateResults(
                char **groups,
                const SelectivityVector &rows,
                const std::vector<VectorPtr> &args,
                bool /* mayPushdown */) override {
                decodedPartial_.decode(*args[0], rows);

                auto baseRowVector = static_cast<const RowVector *>(decodedPartial_.base());
                TIntermediateInput input{baseRowVector};

                rows.applyToSelected([&](auto row) {
                    if (decodedPartial_.is_null_at(row)) {
                        return;
                    }
                    auto decodedIndex = decodedPartial_.index(row);
                    auto *group = groups[row];
                    exec::Aggregate::clearNull(group);
                    input.mergeInto(*accumulator(group), decodedIndex);
                });
            }

            void addSingleGroupRawInput(
                char *group,
                const SelectivityVector &rows,
                const std::vector<VectorPtr> &args,
                bool /* mayPushdown */) override {
                if constexpr (
                    std::is_same_v<TAccumulator, RegrAccumulator> ||
                    std::is_same_v<TAccumulator, ExtendedRegrAccumulator>) {
                    // The args order of linear regression function is (y, x), so we need to
                    // swap the order
                    decodedX_.decode(*args[1], rows);
                    decodedY_.decode(*args[0], rows);
                } else {
                    decodedX_.decode(*args[0], rows);
                    decodedY_.decode(*args[1], rows);
                }

                exec::Aggregate::clearNull(group);
                auto *accumulator = this->accumulator(group);

                rows.applyToSelected([&](auto row) {
                    if (decodedX_.is_null_at(row) || decodedY_.is_null_at(row)) {
                        return;
                    }
                    accumulator->update(decodedX_.value_at<T>(row), decodedY_.value_at<T>(row));
                });
            }

            void addSingleGroupIntermediateResults(
                char *group,
                const SelectivityVector &rows,
                const std::vector<VectorPtr> &args,
                bool /* mayPushdown */) override {
                decodedPartial_.decode(*args[0], rows);

                exec::Aggregate::clearNull(group);
                auto *accumulator = this->accumulator(group);

                auto baseRowVector = static_cast<const RowVector *>(decodedPartial_.base());
                TIntermediateInput input{baseRowVector};

                rows.applyToSelected([&](auto row) {
                    if (decodedPartial_.is_null_at(row)) {
                        return;
                    }
                    auto decodedIndex = decodedPartial_.index(row);
                    input.mergeInto(*accumulator, decodedIndex);
                });
            }

            void extractValues(char **groups, int32_t numGroups, VectorPtr *result)
            override {
                auto vector = (*result)->as<FlatVector<T> >();
                POLLUX_CHECK(vector);
                vector->resize(numGroups);
                uint64_t *rawNulls = getRawNulls(vector);

                T *rawValues = vector->mutableRawValues();
                for (auto i = 0; i < numGroups; ++i) {
                    char *group = groups[i];
                    if (isNull(group)) {
                        vector->set_null(i, true);
                    } else {
                        auto *accumulator = this->accumulator(group);
                        if (TResultAccessor::hasResult(*accumulator)) {
                            clearNull(rawNulls, i);
                            rawValues[i] = static_cast<T>(TResultAccessor::result(*accumulator));
                        } else {
                            vector->set_null(i, true);
                        }
                    }
                }
            }

            void extractAccumulators(char **groups, int32_t numGroups, VectorPtr *result)
            override {
                auto row_vector = (*result)->as<RowVector>();
                row_vector->resize(numGroups);
                for (auto &child: row_vector->children()) {
                    child->resize(numGroups);
                }

                uint64_t *rawNulls = getRawNulls(row_vector);

                TIntermediateResult covarResult{row_vector};

                for (auto i = 0; i < numGroups; ++i) {
                    char *group = groups[i];
                    if (isNull(group)) {
                        row_vector->set_null(i, true);
                    } else {
                        clearNull(rawNulls, i);
                        covarResult.set(i, *accumulator(group));
                    }
                }
            }

        protected:
            void initializeNewGroupsInternal(
                char **groups,
                melon::Range<const vector_size_t *> indices) override {
                setAllNulls(groups, indices);
                for (auto i: indices) {
                    new(groups[i] + offset_) TAccumulator();
                }
            }

        private:
            inline TAccumulator *accumulator(char *group) {
                return exec::Aggregate::value<TAccumulator>(group);
            }

            DecodedVector decodedX_;
            DecodedVector decodedY_;
            DecodedVector decodedPartial_;
        };

        template<
            typename TAccumulator,
            typename TIntermediateInput,
            typename TIntermediateResult,
            typename TResultAccessor>
        exec::AggregateRegistrationResult registerCovariance(
            const std::string &name,
            bool withCompanionFunctions,
            bool overwrite) {
            std::vector<std::shared_ptr<exec::AggregateFunctionSignature> > signatures = {
                // (double, double) -> double
                exec::AggregateFunctionSignatureBuilder()
                .returnType("double")
                .intermediateType(TIntermediateResult::type())
                .argumentType("double")
                .argumentType("double")
                .build(),
                // (real, real) -> real
                exec::AggregateFunctionSignatureBuilder()
                .returnType("real")
                .intermediateType(TIntermediateResult::type())
                .argumentType("real")
                .argumentType("real")
                .build(),
            };

            return exec::registerAggregateFunction(
                name,
                std::move(signatures),
                [](core::AggregationNode::Step step,
                   const std::vector<TypePtr> &argTypes,
                   const TypePtr &resultType,
                   const core::QueryConfig & /*config*/)
            -> std::unique_ptr<exec::Aggregate> {
                    auto rawInputType = exec::isRawInput(step)
                                            ? argTypes[0]
                                            : (exec::isPartialOutput(step) ? DOUBLE() : resultType);
                    switch (rawInputType->kind()) {
                        case TypeKind::DOUBLE:
                            return std::make_unique<CovarianceAggregate<
                                double,
                                TAccumulator,
                                TIntermediateInput,
                                TIntermediateResult,
                                TResultAccessor> >(resultType);
                        case TypeKind::REAL:
                            return std::make_unique<CovarianceAggregate<
                                float,
                                TAccumulator,
                                TIntermediateInput,
                                TIntermediateResult,
                                TResultAccessor> >(resultType);
                        default:
                            POLLUX_UNSUPPORTED(
                                "Unsupported raw input type: {}. Expected DOUBLE or REAL.",
                                rawInputType->toString());
                    }
                },
                withCompanionFunctions,
                overwrite);
        }
    } // namespace

    void registerCovarianceAggregates(
        const std::string &prefix,
        bool withCompanionFunctions,
        bool overwrite) {
        registerCovariance<
            CovarAccumulator,
            CovarIntermediateInput,
            CovarIntermediateResult,
            CovarPopResultAccessor>(
            prefix + kCovarPop, withCompanionFunctions, overwrite);
        registerCovariance<
            CovarAccumulator,
            CovarIntermediateInput,
            CovarIntermediateResult,
            CovarSampResultAccessor>(
            prefix + kCovarSamp, withCompanionFunctions, overwrite);
        registerCovariance<
            CorrAccumulator,
            CorrIntermediateInput,
            CorrIntermediateResult,
            CorrResultAccessor>(prefix + kCorr, withCompanionFunctions, overwrite);
        registerCovariance<
            RegrAccumulator,
            RegrIntermediateInput,
            RegrIntermediateResult,
            RegrInterceptResultAccessor>(
            prefix + kRegrIntercept, withCompanionFunctions, overwrite);
        registerCovariance<
            RegrAccumulator,
            RegrIntermediateInput,
            RegrIntermediateResult,
            RegrSlopeResultAccessor>(
            prefix + kRegrSlop, withCompanionFunctions, overwrite);
        registerCovariance<
            ExtendedRegrAccumulator,
            ExtendedRegrIntermediateInput,
            ExtendedRegrIntermediateResult,
            RegrCountResultAccessor>(
            prefix + kRegrCount, withCompanionFunctions, overwrite);
        registerCovariance<
            ExtendedRegrAccumulator,
            ExtendedRegrIntermediateInput,
            ExtendedRegrIntermediateResult,
            RegrAvgyResultAccessor>(
            prefix + kRegrAvgy, withCompanionFunctions, overwrite);
        registerCovariance<
            ExtendedRegrAccumulator,
            ExtendedRegrIntermediateInput,
            ExtendedRegrIntermediateResult,
            RegrAvgxResultAccessor>(
            prefix + kRegrAvgx, withCompanionFunctions, overwrite);
        registerCovariance<
            ExtendedRegrAccumulator,
            ExtendedRegrIntermediateInput,
            ExtendedRegrIntermediateResult,
            RegrSxyResultAccessor>(
            prefix + kRegrSxy, withCompanionFunctions, overwrite);
        registerCovariance<
            ExtendedRegrAccumulator,
            ExtendedRegrIntermediateInput,
            ExtendedRegrIntermediateResult,
            RegrSxxResultAccessor>(
            prefix + kRegrSxx, withCompanionFunctions, overwrite);
        registerCovariance<
            ExtendedRegrAccumulator,
            ExtendedRegrIntermediateInput,
            ExtendedRegrIntermediateResult,
            RegrSyyResultAccessor>(
            prefix + kRegrSyy, withCompanionFunctions, overwrite);
        registerCovariance<
            ExtendedRegrAccumulator,
            ExtendedRegrIntermediateInput,
            ExtendedRegrIntermediateResult,
            RegrR2ResultAccessor>(
            prefix + kRegrR2, withCompanionFunctions, overwrite);
    }
} // namespace kumo::pollux::aggregate::prestosql
